package com.world.rickcloudendpoint.config;

import com.world.rickcloudendpoint.stream.Task;
import com.world.rickcloudendpoint.stream.consumer.RickConsumer;
import com.world.rickcloudendpoint.stream.consumer.RickConsumerFactory;
import lombok.extern.log4j.Log4j2;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;

import java.io.IOException;

/**
 * @description: 项目配置主人口
 * @author: jlsong
 * @date: 2021/12/04 15:55:27
 * @version: 1.0
 */
@SpringBootApplication
@ComponentScan(basePackages = {"com.world.rickcloudendpoint"})
@MapperScan("com.world.rickcloudendpoint.db.mapper")
@Log4j2
public class RickCloudConfig implements InitializingBean {

    final KafkaConsumerConfig kafkaConsumerConfig;

    final ThreadConfig threadConfig;

    public RickCloudConfig(KafkaConsumerConfig kafkaConsumerConfig, ThreadConfig threadConfig) {
        this.kafkaConsumerConfig = kafkaConsumerConfig;
        this.threadConfig = threadConfig;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 获取kafka消费者配置
        KafkaConsumerConfig kafkaConfig = kafkaConsumerConfig;
        // 调用多线程消费者方案
//        multiThreadedConsumers(kafkaConfig, threadConfig);
    }

    private static void multiThreadedConsumers(KafkaConsumerConfig kafkaConfig, ThreadConfig threadConfig) throws InterruptedException {
        Task consumerTask = new Task(
                threadConfig.getApplicationThreadCount,
                threadConfig.getApplicationThreadCleanupDelay,
                null,
                taskFinished -> {
                    RickConsumer consumer = null;
                    try {
                        //  创建消费者
                        consumer = new RickConsumerFactory(kafkaConfig).buildConsumer();
                        while (true) {
                            // 死循环，如果线程中断，则退出循环
                            if (Thread.interrupted()) {
                                break;
                            }
                            // 不断进行处理
                            consumer.process();
                        }
                    } catch (Exception | Error e) {
                        log.error("Caught exception or error, exiting the application: {}", e.getMessage(), e);
                        System.exit(1);
                    } finally {
                        // 在启动线程前，判断线程是否关闭
                        ensureThreadInterruptStateIsClearedAndClose(consumer);
                        taskFinished.run();
                    }
                });

        log.info("Consumer task created");

        // 任务关闭
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            log.info("Program is going to exit. Have started execution of shutdownHook before this");
            consumerTask.stop();
        }));
        //重新开启任务，直到完成停止
        consumerTask.run().waitForCompletion();
        log.info("Exiting main thread");
    }


    private static void ensureThreadInterruptStateIsClearedAndClose(RickConsumer rickConsumer) {
        Thread.interrupted();
        try {
            rickConsumer.close();
        } catch (IOException e) {
            log.error("Exception on closing firehose consumer", e);
        }
    }
}
